package com.spark.WorCount

import com.alibaba.fastjson.JSON
import org.apache.spark.{SparkConf, SparkContext}

object TopNByScala {
  def main(args: Array[String]): Unit = {
    val sparkConf=new SparkConf()
      .setAppName("TopNByScala")
      .setMaster("local")
    val sc=new SparkContext(sparkConf)
    //1、获取主播的开播记录和用户送礼记录--使用fastjson包解析
    val videoInfoRdd=sc.textFile("datas/video_info.log")
    val giftInfoRdd=sc.textFile("datas/gift_record.log")
    //主播开播记录(video_info.log):主播ID：uid，直播间ID：vid，大区：area  --- (vid,(uid,area))
    //用户送礼记录(gift_record.log)：直播间ID：vid，金币数量：gold ---(vid,gold)
    val videoRdd=videoInfoRdd.map(lines=>{
      val jsonObj=JSON.parseObject(lines)
      val uid=jsonObj.getString("uid")
      val vid=jsonObj.getString("vid")
      val area=jsonObj.getString("area")
      Tuple2(vid, Tuple2(uid, area))
    })
    val giftRdd=giftInfoRdd.map(lines=>{
      val jsonObj=JSON.parseObject(lines)
      val vid = jsonObj.getString("vid")
      val gold = Integer.parseInt(jsonObj.getString("gold"))
      Tuple2(vid, gold)
    })
    //2、对用户送礼记录数据进行聚合，对相同vid的数据求和，
    // 计算主播所在直播间的音浪收入（用户可能在一次直播中给主播送多次礼物） --（vid,gold_sum）,
    val giftReduceRdd=giftRdd.reduceByKey(_ + _)
    //3、把这两份数据join到一块，vid作为join的key--(vid,((uid,area),gold_sum))
    val giftJoinRdd=videoRdd.join(giftReduceRdd)
    //4、使用map迭代join之后的数据，最后获取到uid、area、gold_sum字段-- ((uid,area),gold_sum)
    val giftJoinMap=giftJoinRdd.map(tup=>{
      val uid=tup._2._1._1
      val area=tup._2._1._2
      val gold_sum=tup._2._2
      Tuple2(Tuple2(uid, area), gold_sum)
    })
    //5、使用reduceByKey算子对数据进行聚合（可能存在地区的主播开播多次有多个直播间的音浪收入）-- ((uid,area),gold_sum_all)
    val reduceRDD=giftJoinMap.reduceByKey(_+_)
    //6、接下来对需要使用groupByKey对数据进行分组，所以先使用map进行转换
    // map：(area,(uid,gold_sum_all))
    // groupByKey: area,<(uid,gold_sum_all),(uid,gold_sum_all),(uid,gold_sum_all)
    val groupRdd=reduceRDD.map(tup=>{
      val uid=tup._1._1
      val area=tup._1._2
      val gold_sum_all=tup._2
      Tuple2(area,Tuple2(uid,gold_sum_all))
    }).groupByKey()//分组之后的数据是一个Iterable迭代器
    //7、使用map迭代每个分组内的数据，按照金币数量倒序排序，取前N个，最终输出area,topN
    //这个topN其实就是把前几名主播的id还有金币数量拼接成一个字符串--(area,topN)
    val top3Rdd=groupRdd.map(tup=>{
      val area=tup._1
      val top3=tup._2.toList//把iterable转成list
        .sortBy(_._2)//排序，默认是正序
        .reverse//反转，实现倒序效果
        .take(3)//取前三个元素
        .map(tup=>tup._1+":"+tup._2)//将每一个元素进行转换
        .mkString(",")//使用指定字符把集合转化成字符串（以，分割）
      Tuple2(area,top3)
    })
    //8、使用foreach将结果打印到控制台，多个字段使用制表符分割
    top3Rdd.foreach(tup=>{
      println(tup._1+"\t"+tup._2)
    })

    sc.stop()
  }
}
